Graph of Wikipedia

Project members:

• Vilhelm Agdur, Department of Mathematics, Uppsala University
• Henrik Ekström, Institution
• Simon Johansson, Institution
• Albin Toft, Department of Mathematics, KTH Royal Institute of Technology and Combient Mix AB, Stockholm.

Background

To assume that Wikipedia is a webiste known to most is perhaps not a controversial statement, however to avoid confusion in case someone is not familiar with the website, the following can be found on the Wikipedia article about Wikipedia:

When reading articles on Wikipedia, a reader will be faced with hyper-links to other articles in the Wiki-verse, which might lead a reader onto a path towards a completely different subject than what was on the first page that was read. This phenomena, together with a curiosity about the properties of a graph constructed using the articles and links of Wikipedia lead to this project, with a goal of exploring and analysing what will be reffered to as the "Wiki-Graph". For those unfamiliar with the term "Graph", a graph is a structure amounting to a set of objects in which some pairs of the objects are in some sense "related". The objects correspond to mathematical abstractions called vertices (also called nodes or points) and each of the related pairs of vertices is called an edge. For instance, a graph could be a social network, where the verices are the users and the edges are the friendship relationships among the users. Or as in our case, the vertices could be articles and the edges the hyperlinks between the articles.

Example of a directed graph.

Purpose

The main goal of the project was to explore the Wiki-Graph to answer some exploration oriented questions about the structure. These were questions such as: 1. What is the size of the graph? 2. How dense is it? 3. What can be said about the hierarchical structure of categories? 4. What is the fastest way to go from article A to B, using only the hyper-links? 5. Etc.

Methods

In order to answer these questions we have used the GraphFrames API with Apache Spark. GraphFrames provide tools for analyzing and querying large graphs in a distributed and scalable fashion, with built-in implementations for algorithms such as PageRank, LabelPropagation, BFS and many more.

Data

The data used for the project came from WikiData database dumps. More about precisely what data, how it was ingested, preprocessed and finally joined to produce a GraphFrame, will be presented in upcoming notebooks.

Loading of the Wikipedia data

The data from Wikipedia is available as .sql-file dumps here. So we need to do a little bit of work to get these SQL files into an actual database on the cloud.

For the redirects table, it was small enough to fit into memory on the driver, so we could do it in a fairly simple way. The page and pagelinks tables are too big for that, so we need to be a bit trickier.

As a first step, we download the .sql file:

import java.net.URL
import java.io.File
import org.apache.commons.io.FileUtils

FileUtils.copyURLToFile(new URL("https://dumps.wikimedia.org/enwiki/latest/enwiki-latest-redirect.sql.gz"), new File("/tmp/enwiki-latest-redirect.sql.gz"))

import java.net.URL
import java.io.File
import org.apache.commons.io.FileUtils

Having done this, we first unzip the file, and then move the file from local storage to the DBFS:

gzip -d /tmp/enwiki-latest-redirect.sql.gz

mv file:/tmp/enwiki-latest-redirect.sql /enwiki-latest-redirect.sql

res1: Boolean = true

Having gotten the data onto the DBFS, we can now read it into Spark:

val rawSQLdump = spark.read.textFile("/enwiki-latest-redirect.sql")

rawSQLdump: org.apache.spark.sql.Dataset[String] = [value: string]

The first forty lines are setting up the database, then we get a lot of very long INSERT INTO lines with many many entries being inserted.

println(rawSQLdump.take(40).mkString("\n"))

-- MySQL dump 10.19  Distrib 10.3.34-MariaDB, for debian-linux-gnu (x86_64)
--
-- Host: db1106    Database: enwiki
-- ------------------------------------------------------
-- Server version	10.4.25-MariaDB-log

/*!40101 SET @OLD_CHARACTER_SET_CLIENT=@@CHARACTER_SET_CLIENT */;
/*!40101 SET @OLD_CHARACTER_SET_RESULTS=@@CHARACTER_SET_RESULTS */;
/*!40101 SET @OLD_COLLATION_CONNECTION=@@COLLATION_CONNECTION */;
/*!40101 SET NAMES utf8mb4 */;
/*!40103 SET @OLD_TIME_ZONE=@@TIME_ZONE */;
/*!40103 SET TIME_ZONE='+00:00' */;
/*!40014 SET @OLD_UNIQUE_CHECKS=@@UNIQUE_CHECKS, UNIQUE_CHECKS=0 */;
/*!40014 SET @OLD_FOREIGN_KEY_CHECKS=@@FOREIGN_KEY_CHECKS, FOREIGN_KEY_CHECKS=0 */;
/*!40101 SET @OLD_SQL_MODE=@@SQL_MODE, SQL_MODE='NO_AUTO_VALUE_ON_ZERO' */;
/*!40111 SET @OLD_SQL_NOTES=@@SQL_NOTES, SQL_NOTES=0 */;

--
-- Table structure for table `redirect`
--

DROP TABLE IF EXISTS `redirect`;
/*!40101 SET @saved_cs_client     = @@character_set_client */;
/*!40101 SET character_set_client = utf8 */;
CREATE TABLE `redirect` (
  `rd_from` int(8) unsigned NOT NULL DEFAULT 0,
  `rd_namespace` int(11) NOT NULL DEFAULT 0,
  `rd_title` varbinary(255) NOT NULL DEFAULT '',
  `rd_interwiki` varbinary(32) DEFAULT NULL,
  `rd_fragment` varbinary(255) DEFAULT NULL,
  PRIMARY KEY (`rd_from`),
  KEY `rd_ns_title` (`rd_namespace`,`rd_title`,`rd_from`)
) ENGINE=InnoDB DEFAULT CHARSET=binary ROW_FORMAT=COMPRESSED;
/*!40101 SET character_set_client = @saved_cs_client */;

--
-- Dumping data for table `redirect`
--

/*!40000 ALTER TABLE `redirect` DISABLE KEYS */;

The remaining rows look something like this, except much much longer:

println(rawSQLdump.take(41)(40).substring(0,188) + ",...,"+rawSQLdump.take(41)(40).substring(rawSQLdump.take(41)(40).length()-75, rawSQLdump.take(41)(40).length()))

INSERT INTO `redirect` VALUES (10,0,'Computer_accessibility','',''),(13,0,'History_of_Afghanistan','',''),(14,0,'Geography_of_Afghanistan','',''),(15,0,'Demographics_of_Afghanistan','',''),...,(170997,0,'Supersaturation','',''),(171002,0,'Dubbing','','ADR/post-sync');

Next up, let us strip out the INSERT INTO bit and the initial and final parentheses, then split at each ),(, so that we get each entry as its own string.

val pageDataRows = rawSQLdump.filter(x => x.startsWith("INSERT INTO"))
                             .flatMap(x => x.substring(31, x.length()-2).split("""\),\("""))

pageDataRows: org.apache.spark.sql.Dataset[String] = [value: string]

So now our data looks like this:

println(pageDataRows.take(20).mkString("\n"))

10,0,'Computer_accessibility','',''
13,0,'History_of_Afghanistan','',''
14,0,'Geography_of_Afghanistan','',''
15,0,'Demographics_of_Afghanistan','',''
18,0,'Communications_in_Afghanistan','',''
19,0,'Transport_in_Afghanistan','',''
20,0,'Armed_Forces_of_the_Islamic_Emirate_of_Afghanistan','',''
21,0,'Foreign_relations_of_Afghanistan','',''
23,0,'Assistive_technology','',''
24,0,'Amoeba','',''
25,0,'Autism_spectrum','',''
27,0,'History_of_Albania','',''
29,0,'Demographics_of_Albania','',''
30,0,'As_We_May_Think','',''
35,0,'Politics_of_Albania','',''
36,0,'Economy_of_Albania','',''
40,0,'Afroasiatic_languages','',''
42,0,'Constructed_language','',''
46,0,'Abacus','',''
47,0,'Abalone','',''

This table has a modest amount of rows, only 12.9 million.

pageDataRows.count()

res15: Long = 12937954

The above looks a whole lot like a CSV file, doesn't it? Let's write it to file as such. Note that we write it as text instead of as CSV because our data is in the format of a single string per row.

pageDataRows.toDF().write.mode("overwrite").text("/WikipediaData/enwiki-redirect.csv")

Now we want to read this back in, but with the right schema and column names and so on. So we start by creating the schema. In order to be sure that all the rows got parsed correctly, we add an extra column named _corrupt_record, which will get the raw CSV text whenever it couldn't be parsed right, and otherwise be set to NULL.

import org.apache.spark.sql.types._
// Start by creating a case class of a row entry:
case class WikiRedirect(rd_from:Int,
                        rd_namespace:Int,
                        rd_title:String,
                        rd_interwiki:String,
                        rd_fragment:String
                       )
// then we generate a schema object from the case class: (code copypasted from here: https://sparkbyexamples.com/spark/convert-case-class-to-spark-schema/)
import org.apache.spark.sql.catalyst.ScalaReflection
import org.apache.spark.sql.types.StructType
val pageSchema = ScalaReflection.schemaFor[WikiRedirect].dataType.asInstanceOf[StructType].add("_corrupt_record", StringType, true)

import org.apache.spark.sql.types._
defined class WikiRedirect
import org.apache.spark.sql.catalyst.ScalaReflection
import org.apache.spark.sql.types.StructType
pageSchema: org.apache.spark.sql.types.StructType = StructType(StructField(rd_from,IntegerType,false),StructField(rd_namespace,IntegerType,false),StructField(rd_title,StringType,true),StructField(rd_interwiki,StringType,true),StructField(rd_fragment,StringType,true),StructField(_corrupt_record,StringType,true))

Then we read it back in with the schema we just created:

val readFromCSV = spark.read
                       .options(Map("quote" -> "'", "mode" -> "PERMISSIVE", "columnNameOfCorruptRecord" -> "_corrupt_record"))
                       .schema(pageSchema)
                       .csv("/WikipediaData/enwiki-redirect.csv")

readFromCSV: org.apache.spark.sql.DataFrame = [rd_from: int, rd_namespace: int ... 4 more fields]

Let's have a look at what we just created:

display(readFromCSV)

Next, let us check that we got all the data, and there are no corrupted records:

readFromCSV.createOrReplaceTempView("redirects")

SELECT * FROM redirects WHERE _corrupt_record IS NOT NULL

A single bad row seems fine, we can just drop that one with no harm done.

Let us now write this to the Delta Lake, having filtered out all the bad rows and irrelevant rows, and dropped the columns we don't need. In particular, we remove all redirects to non-main-Wikipedia articles and non-English Wiki articles. (There is exactly one redirect to an article on a different Wiki, and that's on a user talk page.)

SELECT rd_from, rd_title FROM redirects WHERE (rd_from IS NOT NULL) AND (rd_namespace = 0) AND (rd_title IS NOT NULL) AND (rd_interwiki IS NULL) AND (_corrupt_record IS NULL)